Connecting QuestDB using EF

InfluxDB used to be fairly dominant in the timeseries database world, with an exception for system-metrics which are commonly stored in prometheus. But after completely rewriting their database twice (along with the language influxQL -> flux -> SQL-like) and also nuking the capabilities of Influx 3 Core by adding a 72-hour query limit, I do not think that’s true anymore. One timeseries database that has become somewhat popular in the meantime is QuestDB whose free software version does not have limits that alienate most of the user base.

QuestDB uses a PostgreSQL compatible SQL-like interface (several keywords/changes are added because of the nature of timeseries data). This means that you can create tables, insert and query data without any issues if you’re used to working with a SQL database. Creating a database, inserting some data and querying it is as easy as:

CREATE TABLE OrderAnalytics ( 
	Time TIMESTAMP,
	JobOrderId long,
	SensorId VARCHAR,
	MeasurementId VARCHAR,
	Value DOUBLE,
	UnitOfMeasure VARCHAR
) timestamp(Time) PARTITION BY HOUR

INSERT INTO OrderAnalytics
  (Time, JobOrderId, SensorId, MeasurementId, Value, UnitOfMeasure)
VALUES
  (NOW(), '1', 'ESP32-TEMP1', 'Temperature', 10, 'C')

SELECT * FROM OrderAnalytics

As you can see, there are a few parts to the SQL you’ll probably not recognize, but it’s perfectly readable. There are also some nice ways to query the data in a way that automatically makes windows and aggregates fields for you.

SELECT 
  TIME,
  JobOrderId,
  SensorId,
  MeasurementId,
  avg(Value),
  UnitOfMeasure
FROM OrderAnalytics
SAMPLE BY 1h FILL(prev)

Because the SQL interface is PostgreSQL (PGWire) compatible, you can query it using existing libraries. Most of the HolyMate software is written in .NET. You could use their own client library, but you can also use the existing npgsql client. The npgsql client can be connected to Entity Framework Core which is a really nice ORM for C#. QuestDB doesn’t support this directly, but it does work. However, you cannot use the QuestDB specific SQL keywords without manually writing the queries out of the box. This kind of defeats the point of using EF. That is why I spend (more than I intended) some time extending EF Core to support a few of these keywords.

EF Core interface

To connect QuestDB with EF Core, we just need to pretend that its’s a PostgreSQL database that needs a few options disabled. This leads to the following settings:

OrderAnalytics.cs

using System.ComponentModel.DataAnnotations;
using System.ComponentModel.DataAnnotations.Schema;

namespace HolyMate.MES.Domain.Analytics;

public class OrderAnalytics
{
    [Key]
    // It is important to specify the SQL type for the DateTime!
    [Column(TypeName = "timestamp(6)")]
    public DateTime Time { get; set; }
    public long? JobOrderId { get; set; }
    public string? SensorId { get; set; }
    public string? MeasurementId { get; set; }
    public double? Value { get; set; }
    public string? UnitOfMeasure { get; set; }
}

AnalyticsContext.cs

using HolyMate.MES.Domain.Analytics;
using Microsoft.EntityFrameworkCore;
namespace HolyMate.MES.Database;

public class AnalyticsContext(DbContextOptions<AnalyticsContext> options) : DbContext(options)
{
    public DbSet<OrderAnalytics> OrderAnalytics { get; set; } = null!;
}

Appsettings.json

{
    "ConnectionStrings": {
        "Analytics": "Host=127.0.0.1;Port=8812;Database=QuestDB;Username=user;Password=password;ServerCompatibilityMode=NoTypeLoading;"
    },
}

Program.cs

builder.Services
    .AddPooledDbContextFactory<AnalyticsContext>(options =>
    {
        // Its important that the connection string contains `ServerCompatibilityMode=NoTypeLoading;`
        // Because QuestDB doesn't support the type discovery that EF attempts.
        options.UseNpgsql(configuration.GetConnectionString("Analytics"));
        options.EnableDetailedErrors();
        options.EnableSensitiveDataLogging();
    });

And there you go, you can now insert and query models (joining other tables, or creating advanced .Select queries will result in incompatible SQL!). So the following code just works, no problems at all:

private static async Task SaveMeasurement(AnalyticsContext analytics,
    ISA95Context isa95,
    long jobOrderId,
    string sensorId,
    string measurementId,
    DateTime time,
    double value,
    string unitOfMeasure)
{
    var jobOrder = isa95.JobOrders.SingleOrDefault(x => x.Id == jobOrderId);
    if (jobOrder == default)
    {
        throw new GraphQLException($"JobOrder with Id '{jobOrderId}' was not found");
    }
    if (jobOrder.DispatchStatus != DispatchStatus.InProduction)
    {
        throw new GraphQLException($"JobOrder with Id '{jobOrderId}' is not 'InProduction' but '{Enum.GetName(jobOrder.DispatchStatus)}'");
    }

    var analytic = new OrderAnalytics()
    {
        // QuestDB doesn't support timezone conversion (yet), so send it unspecified UTC timestamps!
        Time = DateTime.SpecifyKind(time.ToUniversalTime(), DateTimeKind.Unspecified),
        JobOrderId = jobOrderId,
        MeasurementId = measurementId,
        SensorId = sensorId,
        UnitOfMeasure = unitOfMeasure,
        Value = value
    };

    analytics.OrderAnalytics.Add(analytic);
    await analytics.SaveChangesAsync();
}

[UseProjection]
[UseFiltering]
[UseSorting]
public OrderMeasurements GetAnalyticMeasurements(AnalyticsContext context, long jobOrderId)
{
    var jobMeasurements = context.OrderAnalytics
        .Where(x => x.JobOrderId == jobOrderId);

    if (jobMeasurements.Count() == 0)
    {
        throw new GraphQLException($"There are no measurements for JobOrder '{jobOrderId}'");
    }

    return new OrderMeasurements
    {
        // This are separate queries instead of a big .Select join
        // Since the joining of other tables works slightly different in QuestDB
        EarliestData = jobMeasurements.Min(x => x.Time),
        LatestData = jobMeasurements.Max(x => x.Time),
        MeasurementIds = jobMeasurements.Where(x => x != default).Select(x => x.MeasurementId!).Distinct(),
        SensorIds = jobMeasurements.Where(x => x != default).Select(x => x.SensorId!).Distinct(),
    };
}

However, there is one giant limitation, the windowed queries using the SAMPLE BY t FILL(x) construct is not supported by EF. But almost any timeseries data will eventually need to query it in a windowed fashion. Adding custom SQL to EF’s SQL-Generator is very complex, and mostly undocumented. I spent almost 8 hours reading through EF/ngpsql source code to add this feature. Doing this involved extending a bunch of relevant classes and their factories.

EF core’s SQL generation works by translating LINQ expressions. These expressions are like lambda’s that can be compiled (and inspected/visited) during runtime. This is the code I wrote to add a LINQ Expression for the SampleBy extension

public static class QuestDbLinqExtensions
{
    private static readonly MethodInfo _sampleByMethodInfo =
        typeof(QuestDbLinqExtensions)
            .GetMethod(nameof(SampleBy), BindingFlags.Public | BindingFlags.Static)!;

    private static readonly MethodInfo _efConstantMethod =
        typeof(EF).GetMethod(nameof(EF.Constant), BindingFlags.Public | BindingFlags.Static)!;

    public static IQueryable<T> SampleBy<T>(
        this IQueryable<T> source,
        TimeSpan duration,
        string? fillStrategy = null)
        where T : class
    {
        var durationExpression = Expression.Call(
            _efConstantMethod.MakeGenericMethod(typeof(TimeSpan)),
            Expression.Constant(duration, typeof(TimeSpan))
        );

        var fillExpression = Expression.Call(
            _efConstantMethod.MakeGenericMethod(typeof(string)),
            Expression.Constant(fillStrategy, typeof(string))
        );

        var methodCall = Expression.Call(
            null,
            _sampleByMethodInfo.MakeGenericMethod(typeof(T)),
            source.Expression,
            durationExpression,
            fillExpression
        );

        return source.Provider.CreateQuery<T>(methodCall);
    }
}

This will allow the EF core visitor to walk over the LINQ expression. But currently the method is untranslatable, which will result in a runtime exception. To make the method translatable, we will need to do two things:

• Extend the QueryableMethodTranslatingExpressionVisitor
• Extend the QuerySqlGenerator

The ExpressionVisitor walks over the expression, and can extract the EF.Constant values (our LINQ arguments), and add them as annotations to the query. The SQL generator can then read those annotations at the end, and create the SAMPLE BY section of the query. Both of these are created using a Factory, which will need to be extended as well. The actual factories can be found in git if you want to see those.

This is a part of the ExpressionVisitor, which extends the ngpsl NpgsqlQueryableMethodTranslatingExpressionVisitor class.

protected override Expression VisitMethodCall(MethodCallExpression methodCallExpression)
{
    // This method will be called for _every_ methodCall expression.
    // We only want to translate the SampleBy expressions.
    if (methodCallExpression.Method.IsGenericMethod &&
        methodCallExpression.Method.GetGenericMethodDefinition() == sampleByMethodInfo)
    {
        var source = Visit(methodCallExpression.Arguments[0]);

        if (source is ShapedQueryExpression shapedQuery)
        {
            var selectExpression = (SelectExpression)shapedQuery.QueryExpression;

            // Extract constantExpression values
            var durationExpression = methodCallExpression.Arguments[1];
            var duration = ExtractValue<TimeSpan>(durationExpression);

            var fillStrategy = methodCallExpression.Arguments.Count > 2
                ? ExtractValue<string>(methodCallExpression.Arguments[2])
                : null;

            if (duration == default)
            {
                System.Diagnostics.Debug.WriteLine("WARNING: Duration is default/zero!");
                return shapedQuery;
            }

            var rootSelect = selectExpression;
            while (rootSelect.Tables.Count == 1 && rootSelect.Tables[0] is SelectExpression innerSelect)
            {
                rootSelect = innerSelect;
            }

            // Add annotations for the values.
            rootSelect.AddAnnotation("QuestDB:SampleBy:Duration", duration);
            if (fillStrategy != null)
            {
                rootSelect.AddAnnotation("QuestDB:SampleBy:FillStrategy", fillStrategy);
            }
            return shapedQuery;
        }
    }

    return base.VisitMethodCall(methodCallExpression);
}

In the SQLGenerator, which extends NpgsqlQuerySqlGenerator, we use those annotations

protected override Expression VisitSelect(SelectExpression selectExpression)
{
    var durationAnnotation = selectExpression.FindAnnotation("QuestDB:SampleBy:Duration");
    var fillStrategyAnnotation = selectExpression.FindAnnotation("QuestDB:SampleBy:FillStrategy");

    var result = base.VisitSelect(selectExpression);

    if (durationAnnotation?.Value is TimeSpan duration)
    {
        var fillStrategy = fillStrategyAnnotation?.Value as string;
        AppendSampleBy(duration, fillStrategy);
    }
    else
    {
        System.Diagnostics.Debug.WriteLine("No SAMPLE BY annotation found in SQL generator");
    }

    return result;
}

private void AppendSampleBy(TimeSpan duration, string? fillStrategy)
{
    Sql.AppendLine();
    Sql.Append("SAMPLE BY ");
    Sql.Append(FormatDuration(duration));

    if (!string.IsNullOrEmpty(fillStrategy))
    {
        Sql.Append(" FILL(");
        Sql.Append(fillStrategy);
        Sql.Append(")");
    }
}

And that is all there is to it, now we can generate the following queries using EF Core

[UseProjection]
[UseFiltering]
[UseSorting]
public IQueryable<OrderAnalytics> GetAnalyticValues(AnalyticsContext context, int sampleByMinutes, string? fillStrategy)
{
    return context.OrderAnalytics
        .SampleBy(TimeSpan.FromMinutes(sampleByMinutes), fillStrategy)
        .Select(x => new OrderAnalytics
        {
            Time = x.Time,
            JobOrderId = x.JobOrderId,
            MeasurementId = x.MeasurementId,
            SensorId = x.SensorId,
            Value = context.Avg(x.Value),
            UnitOfMeasure = x.UnitOfMeasure,
        });
}

Which is translated into the following SQL:

SELECT o."Time", o."JobOrderId", o."SensorId", o."MeasurementId", Avg(o."Value") AS "Value", o."UnitOfMeasure"
FROM "OrderAnalytics" AS o
WHERE o."MeasurementId" = @__p_1 AND o."JobOrderId" = @__p_2
SAMPLE BY 64m FILL(NULL)

And it works perfectly. Although it was a lot of research to program, and many non-trivial LINQ queries will still result in invalid SQL, it is much better than hardcoded queries in the code.